package com.at.watermark12.generator;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * 断点式水位线生成器
 *
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/27 22:23
 */
public class MyPuntuatedWatermarkGenerator2<T> implements WatermarkGenerator<T> {
    // 当前时间间戳的延迟值
    private long delayTs;
    // 用来保存，当前为止，最大的事件时间
    private long maxTs;

    public MyPuntuatedWatermarkGenerator2(long delayTs) {
        this.delayTs = delayTs;
        this.maxTs = Long.MIN_VALUE + this.delayTs + 1;
    }

    /**
     * 每条数据来，都会调用一次： 用来提取最大的事件时间，保存下来
     *
     * @param event
     * @param eventTimestamp 提取到的数据的事件时间
     * @param output
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        this.maxTs = Math.max(maxTs, eventTimestamp);
        output.emitWatermark(new Watermark(maxTs - delayTs - 1));
        System.out.println("调用了onEvent方法，获取目前为止的最大时间戳=" + maxTs + ", watermark=" + (maxTs - delayTs - 1));
    }

    /**
     * 周期性调研： 发射 watermark
     *
     * @param output
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTs - delayTs - 1));
        System.out.println("调用了onPeriodicEmit方法，生成watermark=" + (maxTs - delayTs - 1));
    }
}
